package com.dk.mq;

import com.alibaba.fastjson.JSONObject;
import com.dk.entity.DtOrder;
import com.dk.entity.DtUser;
import com.dk.mapper.DtOrderMapper;
import com.dk.mapper.DtUserMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Component
@RocketMQMessageListener(topic = "order_topic",consumerGroup = "dongDongShopConsumer")
public class GradeMq implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private DtOrderMapper dtOrderMapper;

    @Resource
    private DtUserMapper dtUserMapper;

    @Override
    public void onMessage(String s) {
        System.out.println("订单号:"+s);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {//手动接管socketMq
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //获取message
                MessageExt messageExt = list.get(0);
                Integer times = messageExt.getReconsumeTimes();
                try {
                    System.out.println("消息ID >> " + messageExt.getMsgId());
                    //在消费消息之前,先判断此消息有没有消费过，如果消费过,我们就不再消费
                    //解决消费消息的幂等性问题
                    Boolean aBoolean = redisTemplate.boundHashOps("message").hasKey(messageExt.getMsgId());
                    if (!aBoolean){
                        byte[] body = messageExt.getBody();
                        Map map = JSONObject.parseObject(body, Map.class);
                        Object orderIdo =  map.get("orderId");
                        String grade = (String) map.get("grade");

                        //如果是好评就加信誉计分
                        final Long orderId = Long.valueOf((String) orderIdo);
                        DtOrder dtOrder = dtOrderMapper.selectById(orderId);
                        DtUser dtUser = dtUserMapper.selectById(dtOrder.getUserId());
                        if (Objects.equals(grade, "1")){

                            int i = dtUser.getUserLevel() + 1;
                            dtUser.setUserLevel(i);
                            dtUserMapper.updateById(dtUser);
                        }else {
//                            减信誉计分
                            int i = dtUser.getUserLevel() - 1;
                            dtUser.setUserLevel(i);
                            dtUserMapper.updateById(dtUser);
                        }
                        System.out.println("重试次数 >> " + messageExt.getReconsumeTimes());
                        System.out.println("消息主题 >> " + messageExt.getTopic());
//                        //消费成功后将消息放入到redis
//                        redisTemplate.boundHashOps("message").put(messageExt.getMsgId(),phone);
//                        /* 把验证码放在redis中,并设置超时时间 */
//                        redisTemplate.opsForValue().set(phone,numcode,30, TimeUnit.MINUTES);
                    }

                }catch (Exception e){
                    System.out.println("报错,进入错误方法" + e.getMessage());

                    //失败后消息重试,我们可以自己设置死信队列
                    //自定义死信队列
                    if (times > 2){//如果大于2,说明重试三次了，我们可以直接放到自定义的死信队列中
                        System.out.println("重试第 + " + times + " 次");
                        //这个时候我们就可以告诉broker我们消费成功了
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
